package com.kn.keywordmatch.mq;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpStatus;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.kn.keywordmatch.common.RabbitMqCon;
import com.kn.keywordmatch.dto.req.MatchDistanceReq;
import com.kn.keywordmatch.dto.resp.MatchDistanceResp;
import com.kn.keywordmatch.service.MatchService;
import com.kn.keywordmatch.service.PlanService;
import com.kn.keywordmatch.util.ThreadPoolUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

@Component
@Slf4j
public class MatchConsumer {


    @Autowired
    private PlanService planService;
    @Autowired
    private MatchService matchService;

    @Value("${monitor.getMonitorLoca.saveUrl}")
    private String getMonitorLocaSaveUrl;

    @Value("${monitor.getPoleMonitorLoca.saveUrl}")
    private String getPoleMonitorLocaSaveUrl;

    @Value("${monitor.direction.saveUrl}")
    private String getDirectionSaveUrl;

    @Value("${monitor.keyword.saveUrl}")
    private String getKeywordSaveUrl;

    @Autowired
    private RabbitTemplate rabbitTemplate;



    @RabbitListener(queues = RabbitMqCon.HAN_DAN_QUEUE, concurrency = "10")
    public void hanDanReceiveMessage(String message, Channel channel, Message rabbitMessage) {
        try {
            log.info("邯郸监听到消息: " + rabbitMessage.getMessageProperties().getMessageId());

            List<MatchDistanceReq> matchDistanceReqList = JSONObject.parseArray(message, MatchDistanceReq.class);

            // 处理消息逻辑
            ThreadPoolUtil.execute(() -> {
                List<MatchDistanceResp> matchDistanceResp = planService.matchDistance(matchDistanceReqList);
                if(matchDistanceResp.isEmpty()){
                    //log.info("邯郸消费消息匹配为空: {}" , JSONUtil.toJsonStr(matchDistanceReqList));
                    return;
                }
                HttpRequest httpRequest = HttpUtil.createPost(getMonitorLocaSaveUrl);
                httpRequest.contentType("application/json");
                String req = JSON.toJSONString(matchDistanceResp,
                        SerializerFeature.WriteNullNumberAsZero,
                        SerializerFeature.WriteNullListAsEmpty);
                httpRequest.body(req);
                log.warn("邯郸消息推送参数:{} " ,JSONUtil.toJsonStr(req));
                HttpResponse httpResponse = httpRequest.execute();
                String body = httpResponse.body();
                cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
                log.warn("邯郸消息推送返回参数:{} " ,JSONUtil.toJsonStr(body));
                if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
                    // 保存成功状态到
                    log.info("邯郸消费消息成功: " + rabbitMessage.getMessageProperties().getMessageId());
                }else {
                    log.error("邯郸消费消息失败: {}" , JSONUtil.toJsonStr(matchDistanceReqList));
                }
            });
            // 手动 ack
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();

            try {
                channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = RabbitMqCon.XIAO_JI_QUEUE, concurrency = "10")
    public void xiaoJiReceiveMessage(String message, Channel channel, Message rabbitMessage) {

        try {
            log.info("小极监听到消息: " + rabbitMessage.getMessageProperties().getMessageId());

            List<MatchDistanceReq> matchDistanceReqList = JSONObject.parseArray(message, MatchDistanceReq.class);

            ThreadPoolUtil.execute(() -> {
                List<MatchDistanceResp> matchPoleDistanceResp = planService.matchPoleDistance(matchDistanceReqList);
                if(matchPoleDistanceResp.isEmpty()){
                    // log.info("小极消费消息匹配为空: {}" , JSONUtil.toJsonStr(matchDistanceReqList));
                    return;
                }
                // 处理消息逻辑
                HttpRequest httpRequest = HttpUtil.createPost(getPoleMonitorLocaSaveUrl);
                httpRequest.contentType("application/json");
                String req = JSON.toJSONString(matchPoleDistanceResp,
                        SerializerFeature.WriteNullNumberAsZero,
                        SerializerFeature.WriteNullListAsEmpty);
                httpRequest.body(req);
                log.warn("小极消息推送参数:{} " ,JSONUtil.toJsonStr(req));
                HttpResponse httpResponse = httpRequest.execute();
                String body = httpResponse.body();
                cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
                log.warn("小极消息推送返回参数:{} " ,JSONUtil.toJsonStr(body));

                if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
                    // 保存成功状态到
                    log.info("小极消费消息成功: " + rabbitMessage.getMessageProperties().getMessageId());

                }else {
                    log.error("小极消费消息失败: {}" , JSONUtil.toJsonStr(matchDistanceReqList));
                }
            });
            // 手动 ack
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();
            try {
                channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = RabbitMqCon.DIRECTION_QUEUE, concurrency = "10")
    public void directionReceiveMessage(String message, Channel channel, Message rabbitMessage) {

        try {
            log.info("定向监听到消息: " + rabbitMessage.getMessageProperties().getMessageId());

            List<MatchDistanceReq> matchDistanceReqList = JSONObject.parseArray(message, MatchDistanceReq.class);
            if(matchDistanceReqList.isEmpty()){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
           ThreadPoolUtil.execute(() ->{
               long start = System.currentTimeMillis();
               List<MatchDistanceResp> matchPoleDistanceResp = planService.matchByDirection(matchDistanceReqList);
               if(matchPoleDistanceResp.isEmpty()){
                   //log.info("定向消费消息匹配为空: {}" , JSONUtil.toJsonStr(matchDistanceReqList.stream().map(MatchDistanceReq::getDataID).collect(Collectors.toList())));
                   return;
               }
               // 处理消息逻辑
               long end = System.currentTimeMillis();
               HttpRequest httpRequest = HttpUtil.createPost(getDirectionSaveUrl);
               httpRequest.contentType("application/json");
               String req = JSON.toJSONString(matchPoleDistanceResp,
                       SerializerFeature.WriteNullNumberAsZero,
                       SerializerFeature.WriteNullListAsEmpty);
               httpRequest.body(req);
               log.warn("定向消息推送参数:{} " ,JSONUtil.toJsonStr(req));
               HttpResponse httpResponse = httpRequest.execute();
               String body = httpResponse.body();
               cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
               log.warn("定向消息推送返回参数:{} " ,JSONUtil.toJsonStr(body));
               if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
                   log.info("定向消息:directionReceiveMessage 处理完毕耗时：{}",end -start+ "毫秒");
                   log.info("定向消息消费成功: " + rabbitMessage.getMessageProperties().getMessageId());

               }else {
                   log.error("定向消息消费失败: {}" , JSONUtil.toJsonStr(matchDistanceReqList));
               }
           });
            // 手动 ack
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);

        } catch (Exception e) {
            e.printStackTrace();
            try {
                channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = RabbitMqCon.KEYWORD_QUEUE,concurrency = "20")
    public void keywordReceiveMessage(String message, Channel channel, Message rabbitMessage) {

        try {
            List<MatchDistanceReq> matchDistanceReqList = JSONObject.parseArray(message, MatchDistanceReq.class);

            if(matchDistanceReqList.size() == 0){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }
            log.info("关键字监听到消息: " + matchDistanceReqList.stream().map(MatchDistanceReq::getDataID).collect(Collectors.toList()));


            ThreadPoolUtil.execute(() ->
                planService.matchByKeyword(matchDistanceReqList));

            log.info("关键字消息:keywordReceiveMessage 处理完毕：{}",rabbitMessage.getMessageProperties().getDeliveryTag());

            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            e.printStackTrace();

            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息重试次数超过阈值，丢弃消息: " + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
    //@RabbitListener(queues = RabbitMqCon.KEYWORD_EXPRESSION_QUEUE,containerFactory = "factory1")
    public void keywordExpressionReceiveMessage(String message, Channel channel, Message rabbitMessage) {
        try {

            List<MatchDistanceReq> matchDistanceReqList = JSONObject.parseArray(message, MatchDistanceReq.class);
            log.info("关键字表达式监听到消息: " + matchDistanceReqList.size());
            if(matchDistanceReqList.size() == 0){

                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                return;
            }

            ThreadPoolUtil.execute(() ->
                    matchService.matchKeywordByExpression(matchDistanceReqList));


            log.info("关键字表达式消息:keywordReceiveMessage 处理完毕{}",rabbitMessage.getMessageProperties().getDeliveryTag());
            // 手动 ack
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
          } catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息重试次数超过阈值，丢弃消息: " + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    //**********************************测试队列**********************************//
    //
    ////@RabbitListener(queues = RabbitMqCon.KEYWORD_TEST_QUEUE)
    public void testMessage(String message, Channel channel, Message rabbitMessage) {

        log.info("测试队列监听到消息: " + rabbitMessage.getMessageProperties().getMessageId());

        directionReceiveMessage(message, channel, rabbitMessage);
    }

}
